package com.my.study.common;

import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.DescribeTopicsResult;
import org.apache.kafka.clients.admin.NewTopic;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.Collections;
import java.util.concurrent.ExecutionException;

/**
 * 主题配置
 *
 * @author: yidujun
 * @create: 2021/07/22 10:03
 */
@Component("topicConfig")
public class TopicConfig {

    private AdminClient adminClient;

    @Autowired
    public void setAdminClient(AdminClient adminClient) {
        this.adminClient = adminClient;
    }

    /**
     * 创建Topic，分区为2，副本为1
     *
     * @param topicName topic名称
     * @return topic名称
     */
    public String  createTopic(String topicName) {
        return this.createTopic(topicName, 2);
    }

    /**
     * 创建Topic，副本为1
     *
     * @param topicName topic名称
     * @param numPartitions 分区大小
     * @return topic名称
     */
    public String createTopic(String topicName, int numPartitions) {
        return this.createTopic(topicName, numPartitions, (short) 1);
    }

    /**
     * 创建Topic
     *
     * @param topicName  topic名称
     * @param numPartitions 分区大小
     * @param replicationFactor 副本大小
     * @return topic名称
     */
    public String createTopic(String topicName, int numPartitions, short replicationFactor) {
        NewTopic newTopic = new NewTopic(topicName, numPartitions, replicationFactor);
        adminClient.createTopics(Collections.singletonList(newTopic));
        return topicName;
    }

    /**
     * 删除Topic
     * ！！！ 慎用 ！！！
     *
     * @param topicName  topic名称
     * @return topic名称
     */
    public String deleteTopic(String topicName) {
        adminClient.deleteTopics(Collections.singletonList(topicName));
        return topicName;
    }

    /**
     * 查询指定{@code topicName}主题名称的信息
     *
     * @param topicName topic名称
     * @return topic名称
     */
    public String selectTopic(String topicName) {
        DescribeTopicsResult result = adminClient.describeTopics(Collections.singletonList(topicName));
        StringBuilder sb = new StringBuilder("topic信息：");
        try {
            result.all().get().forEach((k, v) -> sb.append("key:").append(k).append(" ").append("value:").append(v));
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }

        return sb.toString();
    }

}
